package com.wpre.vampiremq.mqserver;

import com.wpre.vampiremq.common.Consumer;
import com.wpre.vampiremq.common.MQException;
import com.wpre.vampiremq.mqserver.core.*;
import com.wpre.vampiremq.mqserver.datacenter.DiskDataCenter;
import com.wpre.vampiremq.mqserver.datacenter.MemoryDataCenter;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Created with IntelliJ IDEA.
 * Description:这个类表示虚拟主机,每个虚拟主机管理自己的 交换机 绑定 队列 消息 数据
 * 并且提供api供上层代码调用
 * User: LianBao
 * Date: 2025-01-17
 * Time: 11:56
 */
@Slf4j
public class VirtualHost {
    private String virtualHostName;//虚拟主机名称
    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    private DiskDataCenter diskDataCenter = new DiskDataCenter();
    private Router router = new Router();

    private final Object exchangeLocker = new Object();//操作交换机的锁对象
    private final Object queueLocker = new Object();//操作队列的锁对象

    private ConsumerManager consumerManager = new ConsumerManager(this);

    public String getVirtualHostName() {
        return virtualHostName;
    }

    public MemoryDataCenter getMemoryDataCenter() {
        return memoryDataCenter;
    }

    public DiskDataCenter getDiskDataCenter() {
        return diskDataCenter;
    }


    //提供构造方法,进行初始化操作(建库建表),将内存中的数据恢复到内存中
    public VirtualHost(String virtualHostName) {
        this.virtualHostName = virtualHostName;
        //初始化diskDataCenter
        diskDataCenter.init();
        //加载数据
        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (IOException | ClassNotFoundException e) {
            //业务逻辑的整合者,要对异常进行处理,因为之前都是往上抛,总有人来处理

            log.error("数据恢复到内存中失败!!!");
        }
    }

    // 创建交换机(如果交换机不存在就创建,如果存在就直接返回,创建成功返回true,创建失败返回false)
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType,
                                   boolean durable, boolean autoDelete, Map<String, Object> arguments) {
        //转换交换机名称
        exchangeName = virtualHostName + exchangeName;

        try {
            synchronized (exchangeLocker) {
                //1.判断该交换机是否存在,从内存中查询
                Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                if (exchange != null) {
                    //已经存在
                    return true;
                }
                //2.创建交换机
                Exchange newExchange = new Exchange();
                newExchange.setName(exchangeName);
                newExchange.setDurable(durable);
                newExchange.setType(exchangeType);
                newExchange.setAutoDelete(autoDelete);
                newExchange.setArguments(arguments);
                //3. 把交换机写入硬盘(前提是需要持久化)
                if (durable) {
                    diskDataCenter.insertExchange(newExchange);
                }
                //4.把交换机写入内存
                memoryDataCenter.insertExchange(newExchange);
                log.info("交换机创建成功");
            }
            return true;
        } catch (Exception e) {
            return false;
        }
    }

    //删除交换机
    public boolean exchangeDelete(String exchangeName) {
        exchangeName = virtualHostName + exchangeName;

        try {
            synchronized (exchangeLocker) {
                //从内存中查询
                Exchange toDelete = memoryDataCenter.getExchange(exchangeName);
                if (toDelete == null) {
                    throw new MQException("交换机不存在");
                }
                //删除硬盘上的交换机
                if (toDelete.isDurable()) {
                    diskDataCenter.deleteExchange(exchangeName);
                }
                //删除内存中的交换机
                memoryDataCenter.deleteExchange(exchangeName);
                log.info("交换机删除成功");
            }
            return true;
        } catch (Exception e) {

        }
        return false;
    }

    //创建队列
    public boolean queueDeclare(String queueName, boolean durable, boolean exclusive, boolean autoDelete,
                                Map<String, Object> arguments) {
        queueName = virtualHostName + queueName;

        try {
            synchronized (queueLocker) {
                //从内存中查询,如果存在直接返回
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if (queue != null) {
                    return true;
                }
                //如果不存在,就创建一个
                MSGQueue newQueue = new MSGQueue();
                newQueue.setName(queueName);
                newQueue.setDurable(durable);
                newQueue.setExclusive(exclusive);
                newQueue.setAutoDelete(autoDelete);
                newQueue.setArguments(arguments);

                //在硬盘上创建(如果需要持久化)
                if (durable) {
                    diskDataCenter.insertQueue(newQueue);
                }
                //写入内存
                memoryDataCenter.insertQueue(newQueue);
                log.info("队列创建成功");
            }
            return true;
        } catch (Exception e) {

        }
        return false;
    }

    //删除队列
    public boolean queueDelete(String queueName) {
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker) {
                //从内存中查询
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if (queue == null) {
                    throw new MQException("要删除的队列不存在");
                }
                //硬盘中删除
                if (queue.isDurable()) {
                    diskDataCenter.deleteQueue(queueName);
                }
                //从内存中删除
                memoryDataCenter.deleteQueue(queueName);
            }
            return true;

        } catch (Exception e) {
            return false;
        }
    }

    //创建绑定
    public boolean queueBind(String queueName, String exchangeName, String bindingKey) {
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;

        try {
            synchronized (queueLocker) {
                synchronized (exchangeLocker) {
                    //1.判断绑定是否已经存在
                    Binding binding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (binding != null) {
                        //已经存在了
                        throw new MQException("binding已经存在!!!");
                    }
                    //2.判断bindingKey的合法性
                    if (!router.checkBindingKey(bindingKey)) {
                        throw new MQException("bindingKey不合法");
                    }
                    //创建一个binding对象
                    Binding newBinding = new Binding();
                    newBinding.setExchangeName(exchangeName);
                    newBinding.setQueueName(queueName);
                    newBinding.setBindingKey(bindingKey);
                    //获取对应的交换机和队列,如果对应的队列和交换机不存在,绑定是无法创建的
                    Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                    MSGQueue queue = memoryDataCenter.getQueue(queueName);
                    if (queue == null) {
                        throw new MQException("对应的队列不存在");
                    }
                    if (exchange == null) {
                        throw new MQException("对应的交换机不存在");
                    }

                    //插入硬盘中
                    if (queue.isDurable() && exchange.isDurable()) {
                        diskDataCenter.insertBinding(newBinding);
                    }
                    //插入内存中
                    memoryDataCenter.inertBinding(newBinding);
                }
            }
            return true;
        } catch (Exception e) {
            return false;
        }

    }

    //删除绑定
    public boolean queueUnbind(String queueName, String exchangeName) {
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;

        try {
            synchronized (queueLocker) {
                synchronized (exchangeLocker) {
                    //从内存中查询,看是否存在
                    Binding binding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (binding == null) {
                        throw new MQException("删除绑定失败");
                    }
                    //直接删除:无论是不是需要持久化,都删除
                    //从硬盘中删除
                    memoryDataCenter.deleteBinding(binding);
                    //从内存中删除
                    memoryDataCenter.deleteBinding(binding);
                }
            }
            return true;
        } catch (Exception e) {
            return false;
        }

    }

    //发送消息到指定队列中(发布消息)
    public boolean basicPublish(String exchangeName, String routingKey,
                                BasicProperties basicProperties, byte[] body) {
        try {
            //1.转换交换机名称
            exchangeName = virtualHostName + exchangeName;
            //2.校验routingKey
            if (!router.checkRoutingKey(routingKey)) {
                throw new MQException("RoutingKey不合法");
            }

            //查找交换机
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if (exchange == null) {
                throw new MQException("交换机不存在");
            }
            //判断交换机类型
            if (exchange.getType().equals(ExchangeType.DIRECT)) {
                //直接交换机
                //以routingKey作为队列名,直接把消息写入指定的队列中
                String queueName = virtualHostName + routingKey;//需要构造一下队列名

                //构造消息对象
                Message message = Message.createMessageWithId(routingKey, basicProperties, body);

                //查找队列对象
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if (queue == null) {
                    throw new MQException("队列不存在");
                }
                //队列存在,直接往队列写入消息
                sendMessage(queue, message);

            } else {
                //FANOUT和TOPIC交换机
                //1.找到与该交换机关联的所有绑定
                ConcurrentHashMap<String, Binding> bindings = memoryDataCenter.getBindings(exchangeName);
                for (Map.Entry<String, Binding> entry : bindings.entrySet()) {
                    //遍历所有的绑定对象
                    Binding binding = entry.getValue();
                    //从内存中获取到对应的队列
                    MSGQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
                    if (queue == null) {
                        continue;
                    }

                    //构造消息对象
                    Message message = Message.createMessageWithId(routingKey, basicProperties, body);

                    //判断这个消息是否能转发给队列,如果是FANOUT,所有的队列都要转发,如果是TOPIC,需要判断routingKey和BindingKey是否匹配
                    if (!router.route(exchange.getType(), binding, message)) {
                        continue;
                    }

                    //真正转发消息给队列
                    sendMessage(queue, message);

                }
            }
            return true;
        } catch (Exception e) {
            return false;
        }
    }

    //往队列中写入消息(转发)
    private void sendMessage(MSGQueue queue, Message message) throws IOException, InterruptedException {
        //发送消息,就是把消息写入硬盘和内存
        int deliverMode = message.getDeliverMode();
        // deliverMode为1表示不持久化,为2表示持久化
        if (deliverMode == 2) {
            //写入硬盘
            diskDataCenter.sendMessage(queue, message);
        }
        //写入内存
        memoryDataCenter.sendMessage(message, queue);

        //通知消费者消费消息
        consumerManager.notifyConsume(queue.getName());


    }


    /**
     * 订阅消息
     * 添加一个队列的订阅者(每个队列可能有很多个订阅者),当队列收到消息后就要把消息推送给订阅者
     *
     * @param consumerTag 消费者的身份标识
     * @param queueName   队列名
     * @param autoAck     消息被消费完成后,应答的方式:true为自动应答,false为手动应答(收到调用basicAck方法)
     * @param consumer    回调函数,类型是函数式接口,后续调用basicConsume方法传参时就可以写lambda表达式
     * @return
     */
    public boolean basicConsume(String consumerTag, String queueName, boolean autoAck, Consumer consumer) {
        //构造一个ConsumerEnv对象,把对应的队列找到,再把Consumer对象添加到该队列
        queueName = virtualHostName + queueName;
        try {
            consumerManager.addConsumer(consumerTag, queueName, autoAck, consumer);
        } catch (Exception e) {
            return false;
        }
        return true;
    }

    public boolean basicAck(String queueName, String messageId) {
        queueName = virtualHostName + queueName;
        try {
            //获取到消息和队列
            Message message = memoryDataCenter.getMessage(messageId);
            if (message == null) {
                throw new MQException("消息不存在");
            }
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if (queue == null) {
                throw new MQException("队列不存在");
            }
            //删除硬盘数据
            if (message.getDeliverMode() == 2) {
                diskDataCenter.deleteMessage(queue, message);
            }
            //删除内存中数据
            memoryDataCenter.removeMessage(messageId);
            //删除待确认集合中的消息
            memoryDataCenter.removeMessageWaitAck(queueName, messageId);
            log.info("消息成功确认");
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


}
